package com.example.rabbitmqspringbootdemo;

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.time.LocalDateTime;

@Component
public class AdvancedConsumer {

    /*--------------------消息可靠性发送--------------------*/
    @RabbitListener(queues = "springTransactionQueue")
    public void transactionListener(Message message) {
        System.out.println("[发送者Tansaction模式]received：" + new String(message.getBody()));
    }

    @RabbitListener(queues = "springConfirmQueue")
    public void confirmListener(Message message) {
        System.out.println("[发送者Cofirm模式]received：" + new String(message.getBody()));
    }

    static boolean isFirst = true;

    /*--------------------消息可靠性接收--------------------*/
    @RabbitListener(queues = "springAckQueue")
    public void ackListener(Message message, Channel channel, String msg) {
        //接收消息
        System.out.println("[ACK手动确认]received：" + msg);

        try {
            //处理本地业务
            System.out.println("处理本地业务开始======start======");
            Thread.sleep(2000);
            //模拟本地业务处理出错
            if (msg.contains("5") && isFirst) {
                isFirst = false;
                int i = 1 / 0;  //模拟接收消息出错
            }
            //若没有写代码确认接收，也没有拒绝接收，则消息会一直留在消息队列中unacked
            System.out.println("处理本地业务结束======end======");
            //签收消息
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            e.printStackTrace();
            try {
                System.out.println("开始异常处理......");
                //若出现异常，消息被拒绝后并丢弃(requeue=false)，消息队列中无此消息
                //若出现异常，消息被拒绝后并重回队列(requeue=true)，消息会返回队列，会再次给消费端投递，此时若异常没有解决，就会进入死循环
                channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
            } catch (IOException ioException) {
                ioException.printStackTrace();
            }
        }
    }


    @RabbitListener(queues = "springLimitQueue")
    public void limitListener(Message message, Channel channel, String msg) {
        //接收消息
        System.out.println("[ACK限流]received：" + msg);
        try {
            //只能手动签收消息
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            e.printStackTrace();
            try {
                //若出现异常，消息被拒绝后并丢弃(requeue=false)，消息队列中无此消息
                //若出现异常，消息被拒绝后并重回队列(requeue=true)，消息会返回队列，会再次给消费端投递，此时若异常没有解决，就会进入死循环
                channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
            } catch (IOException ioException) {
                ioException.printStackTrace();
            }
        }
    }

    /*--------------------持久化--------------------*/
    @RabbitListener(queues = "springDurableQueue")
    public void durableistener(Message message) {
        System.out.println("[持久化]received：" + new String(message.getBody()));
    }

    /*--------------------死信--------------------*/
    @RabbitListener(queues = "springNormalQueue")
    public void normalListener(@Payload Integer number, Channel channel, Message message) {
        System.out.println("[死信 Spring NormalConsumer]received:" + number);
        try {
            int i = number / 0;
        } catch (Exception e) {
            try {
                System.out.println("异常消息进入死信队列！");
                //异常消息进入死信队列
                channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
            } catch (IOException ioException) {
                ioException.printStackTrace();
            }
        }
    }

    //死信队列收到死信消息后转发给重定向队列
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @RabbitListener(queues = "springDeadQueue")
    public void deadListener(@Payload Integer number, Channel channel, Message message) {
        try {
            System.out.println("[死信 Spring DeadConsumer]received:" + number);
            //转发给重定向队列
            rabbitTemplate.convertAndSend(
                    "springDeadExchange",
                    "springRedirectRoutingKey",
                    number);
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (IOException ioException) {
            ioException.printStackTrace();
            // System.out.println("异常消息进入死信队列！");
            //异常消息进入死信队列
            //channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
        }
    }

    //重定向队列收到死信消息后处理
    @RabbitListener(queues = "springRedirectQueue")
    public void fromDeadLetter(Integer number, Channel channel, Message message) {
        try {
            System.out.println("[死信 Spring RedirectConsumer]received:" + number);
            // 对应的操作
            int i = number / 2;
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);

    } catch(IOException ioException)
    {
        ioException.printStackTrace();
        // System.out.println("异常消息进入死信队列！");
        //异常消息进入死信队列
        //channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
    }
}
    //--------------------延迟消息队列--------------------
    @RabbitListener(queues = "springDelayQueue")
    public void delayListener(Message message) {
        System.out.println("[延迟消息]received:" + new String(message.getBody()));
        System.out.println("Spring延迟消息接收时间----"+ LocalDateTime.now());
    }
}
